package com.movie.theatre.web.config;

import com.movie.theatre.web.constant.MqEnum;
import com.movie.theatre.web.listener.RouteMessageListenerConcurrently;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Auther: liwu
 * @Date: 2019/12/24 17:12
 * @Description:
 */
@Configuration
@Slf4j
public class RocketMqConfig {

    private String TOPIC = MqEnum.CINEMA_LOCAL_MQ.getTopic();

    private String MQ_GROUP =MqEnum.CINEMA_LOCAL_MQ.getGroup();

    @Value("${rocketmq.consumer.namesrv}")
    private String namesrvAddr;

    @Value("${rocketmq.consumer.accessKey}")
    private String accessKey;

    @Value("${rocketmq.consumer.accessSec}")
    private String accessSec;



    @Bean(destroyMethod = "shutdown")
    public TransactionMQProducer getBaseProducer() throws Exception {

        TransactionMQProducer baseProducer = new TransactionMQProducer(MQ_GROUP,new AclClientRPCHook(new SessionCredentials(accessKey,accessSec)));
        baseProducer.setNamesrvAddr(namesrvAddr);
        baseProducer.setAccessChannel(AccessChannel.CLOUD);
        baseProducer.start();
        return baseProducer;
    }

    /**消费消息*/
    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public DefaultMQPushConsumer mqPushConsumer() throws MQClientException {

        System.setProperty("rocketmq.client.log.loadconfig", "false");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(new AclClientRPCHook(new SessionCredentials(accessKey,accessSec)));
        consumer.setConsumerGroup(MQ_GROUP);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setAccessChannel(AccessChannel.CLOUD);
        //广播消费
        consumer.setMessageModel(MessageModel.BROADCASTING);
        // 订阅该topic下面的某些tag
        consumer.subscribe(TOPIC, "*");

        consumer.registerMessageListener(constructMessageListener());
        return consumer;
    }

    @Bean
    public MessageListenerConcurrently constructMessageListener() {
        return new RouteMessageListenerConcurrently();
    }


}
